package com.yujiahao.bigdata.rdd.persist

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
    TODO 缓存和持久化
      1、如果一个RDD被重复使用，那么也会重头执行数据，并不会从当前操作继续执行
      2、可以将RDD的结果保存起来重复使用，这个操作被叫做持久化操作
        方式一：.cache()内存缓存
          word.cache()
        方式二：.persisit(持久化）
      持久化文件--默认的持久化和cache是一致的，都将数据保存在内存中，但是数据不安全。所以要将数据保存在磁盘中
        word.persist()
      方式三：上面两个一样，这里可以方式二加级别
      word.persist(StorageLevel.DISK_ONLY)
    当前的设置只能将缓存保存在当前应用中使用，在别的应用程序中不能使用
    TODO 设置检查点
      1、如果想要跨越应用，需要采用检查点操作，将数据保存到第三方系统中,
      2、检查点操作为了保证数据执行的正确性，会在执行完成之后，再执行一遍
      3、因此为了提高效率，一般和cache配合使用
        跨应用设置检查点，将数据保存到第三方中
        跨应用设置检查点,一般和cache配合使用
        word.checkpoint()
 */

object Spark01_Persisit {
  def main(args: Array[String]): Unit = {
    //TODO 1、获得Spark环境的连接
    // 参数解读：setAppName("当前执行程序的名字").setMaster("当前的环境")
    // 分布式环境的基本架构为 主 - 从
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)
    //如果想要跨越应用，需要采用检查点操作，将数据保存到第三方系统中,
    // 检查点操作为了保证数据执行的正确性，会在执行完成之后，再执行一遍
    //跨应用设置检查点，将数据保存到第三方中
    sc.setCheckpointDir("cp")

    //TODO 3、业务逻辑
    //如果一个RDD被重复使用，那么也会重头执行数据，并不会从当前操作继续执行
    //可以将RDD的结果保存起来重复使用，，这个操作被叫做持久化操作
    //方式一：.cache()，方式二：.persisit(持久化）。默认的持久化和cache是一致的，都将数据保存在内存中，但是数据不安全。所以要将数据保存在磁盘中
    val rdd: RDD[String] = sc.textFile("data/word.txt")
    val oneWord: RDD[String] = rdd.flatMap(_.split(" "))
    val word: RDD[(String, Int)] = oneWord.map((_, 1))
    //方式一：内存缓存
    word.cache()
    //方式二：持久化文件
    // word.persist()
    //上面两个一样，这里可以方式二加级别
    // word.persist(StorageLevel.DISK_ONLY)

    //跨应用设置检查点,一般和cache配合使用
    word.checkpoint()

    val result: RDD[(String, Int)] = word.reduceByKey(_ + _)
    result.collect().foreach(println)


    println("---------------------------------分割线--------------------------------")
    val value: RDD[(String, Iterable[Int])] = word.groupByKey()
    value.collect() foreach println
    //TODO 2、关闭Spark环境的连接
    sc.stop()

  }

}
